package com.atguigu.extra;

import com.atguigu.bean.UserBehavior;
import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnels;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * 使用bloom过滤器进行一个去重效果
 */
public class FlinkBloomFilter {
    public static void main(String[] args) {
        //来重新实现一下uv
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port",10000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
        env.setParallelism(2);


        DataStreamSource<String> ds = env.readTextFile("input/UserBehavior.csv");
        ds.map(new MapFunction<String, UserBehavior>() {
            @Override
            public UserBehavior map(String value) throws Exception {
                String[] lines = value.split(",");
                UserBehavior userBehavior = new UserBehavior(Long.valueOf(lines[0]),
                        Long.valueOf(lines[1]),
                        Integer.valueOf(lines[2]),
                        String.valueOf(lines[3]),
                        Long.valueOf(lines[4]) *1000);
                return userBehavior;
            }
        })
                //加水印
                .assignTimestampsAndWatermarks(WatermarkStrategy
                //设置最大乱序时间
                .<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner((SerializableTimestampAssigner<UserBehavior>) (element, recordTimestamp) -> element.getTimestamp()))
                //过滤
                .filter((FilterFunction<UserBehavior>) value -> "pv".equals(value.getBehavior()))
                //keyBy聚合
                .keyBy(UserBehavior::getBehavior)
                //如果按照用户id进行keyBy的话，那么半个小时内就会开无数个窗口，每个窗口记录一个用户，
                //虽然这样也可以，但是过于浪费资源，所以不使用

                //开窗
                .window(TumblingEventTimeWindows.of(Time.minutes(30)))
                //处理
                .process(new ProcessWindowFunction<UserBehavior, String, String, TimeWindow>() {
                    @Override
                    public void process(String s,
                                        Context context,
                                        Iterable<UserBehavior> elements,
                                        Collector<String> out) throws Exception {
                        //自定义一个bloom过滤器
                        BloomFilter<Long> bf = BloomFilter.create(Funnels.longFunnel(), 2000000, 0.01);

                        //实现步骤，我们将用户id存储到bloom过滤器中，如果存储成功，表示是第一次存，那么数据不存在，uv+1
                        //如果存储失败，表示已经存过了，uv保持不变

                        //定义一个临时值来存储所有的uv值
                        Long sumUv = 0L;
                        for (UserBehavior element : elements) {
                            boolean isSuccess = bf.put(element.getUserId());
                            if (isSuccess){
                                //表示存储成功，uv+1
                                sumUv++;
                            }
                        }

                        //输出结果
                        out.collect(context.window() + "----->" + sumUv);
                    }
                })
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
